package com.csdj.middleware.config;

import com.csdj.middleware.RocketMqProducer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

@Configuration
public class RocketConfig {

    @Value("${mq.rocket.consumer.group:}")
    private String consumerGroup;
    @Value("${mq.rocket.producer.group:}")
    private String producerGroup;
    @Value("${mq.rocket.name.host:}")
    private String nameServer;
    @Value("${mq.rocket.consumer.topic:test}")
    private String topic;

    @Autowired(required = false)
    private TransactionListener transactionListener;

    private Logger log = LoggerFactory.getLogger("mqconfig");

    @Bean
    @ConditionalOnProperty(prefix = "mq.rocket.consumer",name = "switch",havingValue = "on")
    @ConditionalOnBean(MessageListener.class)
    public DefaultMQPushConsumer consumer(MessageListener msgHandler){
        log.info("启动rocket mqt 消费者 topic:{} host:{}",topic,nameServer);
        if(msgHandler == null){
            System.out.println("msg handler is null");
        }
        final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//        final DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(consumerGroup);

        consumer.setNamesrvAddr(nameServer);
//        consumer.setPullThreadNums(1);
        consumer.setConsumeThreadMin(1);//设置最小线程数
        //以广播的形式接收消息：一条消息会被同一个消费组内所有成员消费
//        consumer.setMessageModel(MessageModel.BROADCASTING);
        try {
            //注册消息监听器
            consumer.registerMessageListener(msgHandler);

            //订阅消费主题   //第二个参数就是tags
            //一个consume如果需要订阅多个topic则可以订阅多次
            consumer.subscribe(topic,"*");
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //监听系统 关闭
        Runtime.getRuntime().addShutdownHook(new Thread(()->{
            consumer.shutdown();
        }));
        return consumer;
    }

    @Bean
    @ConditionalOnProperty(prefix = "mq.rocket.producer",name = "switch",havingValue = "on")
    @ConditionalOnMissingBean(RocketMqProducer.class)
    public RocketMqProducer producer(){
        RocketMqProducer producer = new RocketMqProducer(producerGroup,nameServer,transactionListener);
        producer.setDefaultTopic(topic);
        return producer;
    }
}
